package com.wande.dataplatform.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.DataLineage;
import com.wande.dataplatform.domain.DataSource;
import com.wande.dataplatform.domain.EtlTask;
import com.wande.dataplatform.domain.Metadata;
import com.wande.dataplatform.domain.dto.ImpactAnalysisDTO;
import com.wande.dataplatform.domain.dto.MetadataQueryDTO;
import com.wande.dataplatform.domain.vo.DataLineageVO;
import com.wande.dataplatform.domain.vo.MetadataVO;
import com.wande.dataplatform.mapper.DataLineageMapper;
import com.wande.dataplatform.mapper.DataSourceMapper;
import com.wande.dataplatform.mapper.EtlTaskMapper;
import com.wande.dataplatform.mapper.MetadataMapper;
import com.wande.dataplatform.service.IMetadataService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 元数据服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MetadataServiceImpl implements IMetadataService {

    private final MetadataMapper metadataMapper;
    private final DataLineageMapper lineageMapper;
    private final DataSourceMapper dataSourceMapper;
    private final EtlTaskMapper etlTaskMapper;
    private final DataSourceServiceImpl dataSourceService;

    @Override
    public TableDataInfo<MetadataVO> queryMetadataList(MetadataQueryDTO queryDTO, PageQuery pageQuery) {
        Metadata metadata = new Metadata();
        if (queryDTO != null) {
            metadata.setDatasourceId(queryDTO.getDatasourceId());
            metadata.setDatabaseName(queryDTO.getDatabaseName());
            metadata.setTableName(queryDTO.getTableName());
            metadata.setTableType(queryDTO.getTableType());
            metadata.setProjectId(queryDTO.getProjectId());
            metadata.setStatus(queryDTO.getStatus());
        }

        Page<MetadataVO> page = pageQuery.build();
        List<MetadataVO> list = metadataMapper.selectMetadataVOList(metadata);
        
        // Format data size and parse JSON
        list.forEach(vo -> {
            if (vo.getDataSize() != null) {
                vo.setDataSizeFormatted(formatDataSize(vo.getDataSize()));
            }
            // Parse fields from JSON
            if (StrUtil.isNotBlank(vo.getFieldsJson())) {
                try {
                    vo.setFields(JSONUtil.toList(vo.getFieldsJson(), MetadataVO.FieldInfo.class));
                } catch (Exception e) {
                    log.warn("Failed to parse fields JSON: {}", e.getMessage());
                }
            }
            // Parse indexes from JSON
            if (StrUtil.isNotBlank(vo.getIndexesJson())) {
                try {
                    vo.setIndexes(JSONUtil.toList(vo.getIndexesJson(), MetadataVO.IndexInfo.class));
                } catch (Exception e) {
                    log.warn("Failed to parse indexes JSON: {}", e.getMessage());
                }
            }
        });

        page.setRecords(list);
        page.setTotal(list.size());
        return TableDataInfo.build(page);
    }

    @Override
    public MetadataVO getMetadataById(Long id) {
        MetadataVO vo = metadataMapper.selectMetadataVOById(id);
        if (vo == null) {
            throw new ServiceException("元数据不存在");
        }
        
        if (vo.getDataSize() != null) {
            vo.setDataSizeFormatted(formatDataSize(vo.getDataSize()));
        }
        
        // Parse fields from JSON
        if (StrUtil.isNotBlank(vo.getFieldsJson())) {
            try {
                vo.setFields(JSONUtil.toList(vo.getFieldsJson(), MetadataVO.FieldInfo.class));
            } catch (Exception e) {
                log.warn("Failed to parse fields JSON: {}", e.getMessage());
            }
        }
        
        // Parse indexes from JSON
        if (StrUtil.isNotBlank(vo.getIndexesJson())) {
            try {
                vo.setIndexes(JSONUtil.toList(vo.getIndexesJson(), MetadataVO.IndexInfo.class));
            } catch (Exception e) {
                log.warn("Failed to parse indexes JSON: {}", e.getMessage());
            }
        }
        
        return vo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Integer collectMetadata(Long datasourceId) {
        DataSource dataSource = dataSourceMapper.selectById(datasourceId);
        if (dataSource == null) {
            throw new ServiceException(DataPlatformErrorCode.DATASOURCE_NOT_FOUND.getMessage());
        }

        int count = 0;
        try (Connection conn = dataSourceService.getConnection(datasourceId)) {
            DatabaseMetaData dbMetaData = conn.getMetaData();
            
            // Get all tables
            try (ResultSet tables = dbMetaData.getTables(null, null, "%", new String[]{"TABLE", "VIEW"})) {
                while (tables.next()) {
                    String tableName = tables.getString("TABLE_NAME");
                    String tableType = tables.getString("TABLE_TYPE");
                    String remarks = tables.getString("REMARKS");
                    
                    // Collect table metadata
                    Metadata metadata = collectSingleTable(datasourceId, conn, tableName, tableType, remarks);
                    if (metadata != null) {
                        count++;
                    }
                }
            }
        } catch (Exception e) {
            log.error("采集元数据失败: {}", e.getMessage(), e);
            throw new ServiceException("采集元数据失败: " + e.getMessage());
        }

        return count;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long collectTableMetadata(Long datasourceId, String databaseName, String tableName) {
        DataSource dataSource = dataSourceMapper.selectById(datasourceId);
        if (dataSource == null) {
            throw new ServiceException(DataPlatformErrorCode.DATASOURCE_NOT_FOUND.getMessage());
        }

        try (Connection conn = dataSourceService.getConnection(datasourceId)) {
            DatabaseMetaData dbMetaData = conn.getMetaData();
            
            // Get table info
            try (ResultSet tables = dbMetaData.getTables(null, databaseName, tableName, new String[]{"TABLE", "VIEW"})) {
                if (tables.next()) {
                    String tableType = tables.getString("TABLE_TYPE");
                    String remarks = tables.getString("REMARKS");
                    
                    Metadata metadata = collectSingleTable(datasourceId, conn, tableName, tableType, remarks);
                    return metadata != null ? metadata.getId() : null;
                }
            }
        } catch (Exception e) {
            log.error("采集表元数据失败: {}", e.getMessage(), e);
            throw new ServiceException("采集表元数据失败: " + e.getMessage());
        }

        return null;
    }

    private Metadata collectSingleTable(Long datasourceId, Connection conn, String tableName, 
                                       String tableType, String remarks) throws Exception {
        DatabaseMetaData dbMetaData = conn.getMetaData();
        
        // Collect fields
        List<MetadataVO.FieldInfo> fields = new ArrayList<>();
        try (ResultSet columns = dbMetaData.getColumns(null, null, tableName, "%")) {
            while (columns.next()) {
                MetadataVO.FieldInfo field = new MetadataVO.FieldInfo();
                field.setFieldName(columns.getString("COLUMN_NAME"));
                field.setFieldType(columns.getString("TYPE_NAME"));
                field.setFieldComment(columns.getString("REMARKS"));
                field.setNullable("YES".equals(columns.getString("IS_NULLABLE")));
                field.setDefaultValue(columns.getString("COLUMN_DEF"));
                fields.add(field);
            }
        }
        
        // Collect primary keys
        Set<String> primaryKeys = new HashSet<>();
        try (ResultSet pks = dbMetaData.getPrimaryKeys(null, null, tableName)) {
            while (pks.next()) {
                primaryKeys.add(pks.getString("COLUMN_NAME"));
            }
        }
        
        // Mark primary key fields
        fields.forEach(field -> field.setPrimaryKey(primaryKeys.contains(field.getFieldName())));
        
        // Collect indexes
        List<MetadataVO.IndexInfo> indexes = new ArrayList<>();
        try (ResultSet indexInfo = dbMetaData.getIndexInfo(null, null, tableName, false, false)) {
            Map<String, MetadataVO.IndexInfo> indexMap = new HashMap<>();
            while (indexInfo.next()) {
                String indexName = indexInfo.getString("INDEX_NAME");
                if (indexName != null) {
                    indexMap.computeIfAbsent(indexName, k -> {
                        MetadataVO.IndexInfo idx = new MetadataVO.IndexInfo();
                        idx.setIndexName(indexName);
                        try {
                            idx.setUnique(!indexInfo.getBoolean("NON_UNIQUE"));
                        } catch (Exception e) {
                            idx.setUnique(false);
                        }
                        idx.setColumns(new ArrayList<>());
                        return idx;
                    });
                    MetadataVO.IndexInfo index = indexMap.get(indexName);
                    index.getColumns().add(indexInfo.getString("COLUMN_NAME"));
                }
            }
            indexes.addAll(indexMap.values());
        }
        
        // Check if metadata exists
        LambdaQueryWrapper<Metadata> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(Metadata::getDatasourceId, datasourceId)
               .eq(Metadata::getTableName, tableName);
        Metadata existing = metadataMapper.selectOne(wrapper);
        
        Metadata metadata = existing != null ? existing : new Metadata();
        metadata.setDatasourceId(datasourceId);
        metadata.setTableName(tableName);
        metadata.setTableComment(remarks);
        metadata.setTableType(tableType);
        metadata.setFields(JSONUtil.toJsonStr(fields));
        metadata.setIndexes(JSONUtil.toJsonStr(indexes));
        metadata.setLastCollectTime(LocalDateTime.now());
        metadata.setStatus(1);
        
        if (existing != null) {
            metadataMapper.updateById(metadata);
        } else {
            metadataMapper.insert(metadata);
        }
        
        return metadata;
    }

    @Override
    public TableDataInfo<MetadataVO> searchMetadata(String keyword, PageQuery pageQuery) {
        LambdaQueryWrapper<Metadata> wrapper = new LambdaQueryWrapper<>();
        wrapper.and(w -> w.like(Metadata::getTableName, keyword)
                         .or()
                         .like(Metadata::getTableComment, keyword));
        wrapper.eq(Metadata::getStatus, 1);
        wrapper.orderByDesc(Metadata::getUpdateTime);
        
        Page<Metadata> page = metadataMapper.selectPage(pageQuery.build(), wrapper);
        
        List<MetadataVO> voList = page.getRecords().stream().map(metadata -> {
            MetadataVO vo = metadataMapper.selectMetadataVOById(metadata.getId());
            if (vo.getDataSize() != null) {
                vo.setDataSizeFormatted(formatDataSize(vo.getDataSize()));
            }
            return vo;
        }).collect(Collectors.toList());
        
        Page<MetadataVO> voPage = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
        voPage.setRecords(voList);
        
        return TableDataInfo.build(voPage);
    }

    @Override
    public List<MetadataVO> exportMetadata(MetadataQueryDTO queryDTO) {
        Metadata metadata = new Metadata();
        if (queryDTO != null) {
            metadata.setDatasourceId(queryDTO.getDatasourceId());
            metadata.setDatabaseName(queryDTO.getDatabaseName());
            metadata.setTableName(queryDTO.getTableName());
            metadata.setTableType(queryDTO.getTableType());
            metadata.setProjectId(queryDTO.getProjectId());
            metadata.setStatus(queryDTO.getStatus());
        }
        
        return metadataMapper.selectMetadataVOList(metadata);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Integer importMetadata(List<MetadataVO> metadataList) {
        if (CollUtil.isEmpty(metadataList)) {
            return 0;
        }
        
        int count = 0;
        for (MetadataVO vo : metadataList) {
            Metadata metadata = new Metadata();
            metadata.setDatasourceId(vo.getDatasourceId());
            metadata.setDatabaseName(vo.getDatabaseName());
            metadata.setTableName(vo.getTableName());
            metadata.setTableComment(vo.getTableComment());
            metadata.setTableType(vo.getTableType());
            metadata.setFields(JSONUtil.toJsonStr(vo.getFields()));
            metadata.setIndexes(JSONUtil.toJsonStr(vo.getIndexes()));
            metadata.setDataSize(vo.getDataSize());
            metadata.setRowCount(vo.getRowCount());
            metadata.setProjectId(vo.getProjectId());
            metadata.setStatus(1);
            
            metadataMapper.insert(metadata);
            count++;
        }
        
        return count;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteMetadata(Long id) {
        return metadataMapper.deleteById(id) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean batchDeleteMetadata(List<Long> ids) {
        if (CollUtil.isEmpty(ids)) {
            return false;
        }
        return metadataMapper.deleteBatchIds(ids) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean buildLineage(Long taskId, String taskType) {
        if ("ETL_TASK".equals(taskType)) {
            return buildEtlLineage(taskId);
        }
        return false;
    }

    private Boolean buildEtlLineage(Long taskId) {
        EtlTask etlTask = etlTaskMapper.selectById(taskId);
        if (etlTask == null) {
            return false;
        }
        
        // Find source and target metadata
        Long sourceMetadataId = findMetadataByDatasource(etlTask.getSourceId());
        Long targetMetadataId = findMetadataByDatasource(etlTask.getTargetId());
        
        if (sourceMetadataId == null || targetMetadataId == null) {
            return false;
        }
        
        // Create lineage
        DataLineage lineage = new DataLineage();
        lineage.setSourceTableId(sourceMetadataId);
        lineage.setTargetTableId(targetMetadataId);
        lineage.setLineageType("ETL");
        lineage.setTaskId(taskId);
        lineage.setTaskType("ETL_TASK");
        lineage.setProjectId(etlTask.getProjectId());
        lineage.setStatus(1);
        
        return lineageMapper.insert(lineage) > 0;
    }

    private Long findMetadataByDatasource(Long datasourceId) {
        LambdaQueryWrapper<Metadata> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(Metadata::getDatasourceId, datasourceId);
        wrapper.last("LIMIT 1");
        Metadata metadata = metadataMapper.selectOne(wrapper);
        return metadata != null ? metadata.getId() : null;
    }

    @Override
    public List<DataLineageVO> getUpstream(Long tableId) {
        return lineageMapper.selectUpstreamByTableId(tableId);
    }

    @Override
    public List<DataLineageVO> getDownstream(Long tableId) {
        return lineageMapper.selectDownstreamByTableId(tableId);
    }

    @Override
    public ImpactAnalysisDTO analyzeImpact(Long tableId, String changeType) {
        Metadata metadata = metadataMapper.selectById(tableId);
        if (metadata == null) {
            throw new ServiceException("元数据不存在");
        }
        
        ImpactAnalysisDTO result = new ImpactAnalysisDTO();
        result.setTableId(tableId);
        result.setTableName(metadata.getTableName());
        result.setChangeType(changeType);
        
        // Get upstream and downstream
        List<DataLineageVO> upstream = getUpstream(tableId);
        List<DataLineageVO> downstream = getDownstream(tableId);
        
        result.setUpstreamCount(upstream.size());
        result.setDownstreamCount(downstream.size());
        
        // Analyze affected tasks
        List<ImpactAnalysisDTO.AffectedTask> affectedTasks = new ArrayList<>();
        downstream.forEach(lineage -> {
            if (lineage.getTaskId() != null) {
                ImpactAnalysisDTO.AffectedTask task = new ImpactAnalysisDTO.AffectedTask();
                task.setTaskId(lineage.getTaskId());
                task.setTaskName(lineage.getTaskName());
                task.setTaskType(lineage.getTaskType());
                task.setImpactType("DATA_DEPENDENCY");
                affectedTasks.add(task);
            }
        });
        result.setAffectedTasks(affectedTasks);
        
        // Determine impact level
        int totalImpact = downstream.size() + affectedTasks.size();
        if (totalImpact == 0) {
            result.setImpactLevel("LOW");
            result.setImpactDescription("无下游依赖，影响范围小");
        } else if (totalImpact <= 5) {
            result.setImpactLevel("MEDIUM");
            result.setImpactDescription("存在少量下游依赖，需要评估影响");
        } else {
            result.setImpactLevel("HIGH");
            result.setImpactDescription("存在大量下游依赖，需要谨慎变更");
        }
        
        return result;
    }

    @Override
    public Object getLineageGraph(Long tableId, Integer depth) {
        if (depth == null || depth <= 0) {
            depth = 2;
        }
        
        Map<String, Object> graph = new HashMap<>();
        List<Map<String, Object>> nodes = new ArrayList<>();
        List<Map<String, Object>> edges = new ArrayList<>();
        
        // Build graph recursively
        Set<Long> visited = new HashSet<>();
        buildGraphRecursive(tableId, depth, 0, nodes, edges, visited, true);
        buildGraphRecursive(tableId, depth, 0, nodes, edges, visited, false);
        
        graph.put("nodes", nodes);
        graph.put("edges", edges);
        
        return graph;
    }

    private void buildGraphRecursive(Long tableId, int maxDepth, int currentDepth,
                                     List<Map<String, Object>> nodes,
                                     List<Map<String, Object>> edges,
                                     Set<Long> visited, boolean upstream) {
        if (currentDepth >= maxDepth || visited.contains(tableId)) {
            return;
        }
        
        visited.add(tableId);
        
        // Add current node
        Metadata metadata = metadataMapper.selectById(tableId);
        if (metadata != null) {
            Map<String, Object> node = new HashMap<>();
            node.put("id", tableId);
            node.put("label", metadata.getTableName());
            node.put("type", metadata.getTableType());
            nodes.add(node);
        }
        
        // Get related lineages
        List<DataLineageVO> lineages = upstream ? getUpstream(tableId) : getDownstream(tableId);
        
        for (DataLineageVO lineage : lineages) {
            Long relatedTableId = upstream ? lineage.getSourceTableId() : lineage.getTargetTableId();
            
            // Add edge
            Map<String, Object> edge = new HashMap<>();
            edge.put("source", upstream ? relatedTableId : tableId);
            edge.put("target", upstream ? tableId : relatedTableId);
            edge.put("label", lineage.getLineageType());
            edges.add(edge);
            
            // Recursive call
            buildGraphRecursive(relatedTableId, maxDepth, currentDepth + 1, nodes, edges, visited, upstream);
        }
    }

    private String formatDataSize(Long bytes) {
        if (bytes == null || bytes == 0) {
            return "0 B";
        }
        
        String[] units = {"B", "KB", "MB", "GB", "TB"};
        int unitIndex = 0;
        double size = bytes.doubleValue();
        
        while (size >= 1024 && unitIndex < units.length - 1) {
            size /= 1024;
            unitIndex++;
        }
        
        return String.format("%.2f %s", size, units[unitIndex]);
    }
}
